package com.ydl.learning.flink.demo.pipeline

import com.ydl.learning.flink.demo.pipeline.java.entity.JobConfig

/**
 *
 *
 * @author ydl
 * @since 2022/3/27
 */
class Builder2 {

  private[this] var _jobConfig: JobConfig = null

  def jobConfig: JobConfig = _jobConfig

  private def jobConfig_=(value: JobConfig): Unit = {
    _jobConfig = value
  }


  private[this] var _init: Source[PipelineData] = null

  def source: Source[PipelineData] = _init

  private def source_=(value: Source[PipelineData]): Unit = {
    _init = value
  }


  private[this] var _processes: List[Process[PipelineData]] = List[Process[PipelineData]]()

  def processes: List[Process[PipelineData]] = _processes

  private def processes_=(value: List[Process[PipelineData]]): Unit = {
    _processes = value
  }


  //TODO 测流输出
  private var processSideOut: String = null

  //TODO 监控
  private var processMonitor = null

  private[this] var _sink: Sink[PipelineData] = null

  def sink: Sink[PipelineData] = _sink

  private def sink_=(value: Sink[PipelineData]): Unit = {
    _sink = value
  }


  //TODO 流克隆
  private var cloneStream = null
  //TODO 克隆流处理
  private var cloneStreamProcess = null
  //TODO 克隆流输出
  private var cloneStreamSink = null

  def jobConfig(jobConfig: JobConfig): Builder2 = {
    this.jobConfig = jobConfig
    this
  }

  def source(source: Source[PipelineData]): Builder2 = {
    this.source = source
    this
  }

  def process(process: Process[PipelineData]): Builder2 = {
    this.processes = this.processes :+ process
    this
  }

  def sink(sink: Sink[PipelineData]): Builder2 = {
    this.sink = sink
    this
  }

  def build = new Pipeline2(this)
}
